# TOOLS
def remove_none_label(l: dict) -> dict:
# remove None label in labels
mark = []
for k, v in l.items():
if v == None:
mark.append(k)
for m in mark:
l.pop(m)
return l
def get_node_centric_graph(df, node_name):
"""
:df: data frame
:node_name: the center node name
"""
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']
sources, targets, relations = [], [], []
for i in range(len(df['source'])):
if df.loc[i]['source'] == node_name:
sources.append(node_name)
targets.append(df.loc[i]['target'])
relations.append(df.loc[i]['edge'])
elif isinstance(df.loc[i]['source'], tuple):
if df.loc[i]['source'][0] == node_name:
sources.append(df.loc[i]['source'])
targets.append(df.loc[i]['target'])
relations.append(df.loc[i]['edge'])
new_df = pd.DataFrame({'source': sources, 'target': targets, 'edge': relations})
return new_df
def draw_graph(df):
"""
draw graph given a triple data frame
"""
# create a directed-graph from a dataframe
G=nx.from_pandas_edgelist(t, source = 'source', target = 'target', edge_attr = 'edge', create_using= nx.DiGraph())
plt.figure(figsize=(100,100))
pos = nx.spring_layout(G)
edge_labels = remove_none_label(nx.get_edge_attributes(G, 'edge'))
# remove coordinator nodes
node_labels = {n: n for n in G if not isinstance(n, tuple)}
# remove coordinator nodes
node_list = [i for i in list(G) if not isinstance(i, tuple)]
# draw nodes
nx.draw_networkx_nodes(G, pos, nodelist=node_list)
# draw all edges
nx.draw_networkx_edges(G, pos)
# draw labels for all desired nodes
nx.draw_networkx_labels(G, pos, labels=node_labels)
# draw labels for all desired node-node edges
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)
plt.show()
return
